package com.myself.producer;

import com.myself.partitions.MyPatitions;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @author W.S.
 * @version v1.0.0
 * @description kafka 生产者
 * @date Created in 2019/4/23 14:40
 */
public class ProducerDemo {
    public static void main(String[] args) {
        // 初始化链接配置信息
        Properties config = new Properties();
        // 链接ip
        config.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        // key 序列化
        config.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // value 序列化
        config.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 设置自定义的分区管理器
        // config.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIGz, MyPatitions.class.getName());
        String topicName = "my-topic";
        // key来区分消息从发到那个分区 如果key为null则消息依照轮训的方式发送 否则通过key值的hash值来适配分区
        String key = null;
        String value = "我是一个value2";
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config);
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topicName,key,value);
        /**
         * 生产者发送一条消息-----》序列化器------》分区管理器（可自定义规则）-------》分区----》broker
         */
        Future<RecordMetadata> send =  producer.send(record);
        producer.flush();
        producer.close();
        System.out.println("发送完毕-------");
    }
}
